package SparkRDD.RDD算子.Transformations.过滤操作

import org.apache.commons.lang.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

class IP统计清洗 {

  @Test
  def IPTest: Unit ={

    // 1.创建SparkContext
    val conf = new SparkConf().setAppName("IP_count")
    val sc   = new SparkContext(conf)

    // 2.读取数据文件，生成数据集
    val sourceRDD = sc.textFile("文件路径")

    // 3.取出ID，赋予词频
    val idRDD = sourceRDD.map(item=>(item.split(" ")(0),1))

    // 4.简单清洗
    //    去掉空数据
    //    去掉非法数据
    //    根据业务规整数据
    val cleanRDD = idRDD.filter(item => StringUtils.isNotEmpty(item._1))

    // 5.根据IP出现次数聚合
    val aggRDD = cleanRDD.reduceByKey((curr,agg) => curr+agg)

    // 6.排序
    val sortRDD = aggRDD.sortBy(item => item._2,ascending = false)

    // 取出并打印结果
    sortRDD.take(10).foreach(item => println(item))
  }

}
